package com.源码分析;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import org.junit.Test;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

/**一个bug的演示：问题出在subscribeOn(...)后面紧接一个publishOn(...)的时候
 * @author Ziroom
 *
 */
public class ThreadDemo4 {

	@Test
	// Flux的publishOn
	public void testSyncToAsync() throws InterruptedException {
		final Random random = new Random();
	    Flux.create(sink -> {
	        List<Integer> list = new ArrayList<>();
	        Integer i=0;
	        while(list.size() != 100) {
		        int value = random.nextInt(1000);
		        list.add(value);
		        i=i+1;
		        System.out.println("所发射元素产生的线程：" + Thread.currentThread().getName() + " num: " + i);
		        sink.next(value);
		        try {
		            Thread.sleep(10);
		        } catch (InterruptedException e) {
		            e.printStackTrace();
		        }
	        }
	        sink.complete();
	    })
	    // subscribeOn(...)后面跟一个publishOn(...)，会出现请求完2个元素后就不再请求元素了，必须等create全部生成完，才开始再次请求元素，
	    // 这是因为FluxSubscribeOn$SubscribeOnSubscriber的requestUpstream(...)方法：会进入worker.schedule(() -> s.request(n))，但由于worker的exec是一个单工作线程的线程池，
	    // 所以导致s.request(n)一直得不到执行，因为请求元素和下发元素都在subscribeOn(...)的线程里，导致下发元素不完成，就无法执行请求元素。
	    // 把subscribeOn(...)改为.subscribeOn(Schedulers.elastic(), false)可以解决该问题
	    .subscribeOn(Schedulers.elastic())
	    .publishOn(Schedulers.elastic(), 2)
	    .subscribe(x -> {
	    	try {
	            Thread.sleep(50);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        }
	    	System.out.println("所消费元素的线程：" + Thread.currentThread().getName() + " num: " + x);
	    });
	    Thread.sleep(3000000);
	}

}
